package com.shujia.tf

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo8SideOut {
  def main(args: Array[String]): Unit = {
    /**
      * 将一个流拆分成多个流
      *
      */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
      * 将男和女分成两个流
      *
      */

    /**
      * 1、使用filter进行拆分
      *
      */

    val filterFDS: DataStream[String] = studentDS.filter(line => "男".equals(line.split(",")(3)))

    val filterMDS: DataStream[String] = studentDS.filter(line => "女".equals(line.split(",")(3)))


    //filterFDS.print()
    //filterMDS.print()

    /**
      * 1、side out
      *
      * 再主的流上从侧边输出一个，
      * filter 是对源ds使用的多次，每一次都需要全量比较，性能较差
      * site out 只需要比较一次即可
      *
      */
    val f: OutputTag[String] = OutputTag[String]("男")
    val m: OutputTag[String] = OutputTag[String]("女")


    val sideOutDS: DataStream[String] = studentDS.process(new ProcessFunction[String, String] {
      /**
        * process： 将ds的数据一条一条传递给processElement，
        * 再方法内多数据做处理，处理完可以发生堕胎数据到下游
        *
        * @param line : 一行数据
        * @param ctx  : 上下文对象
        * @param out  : 用于将数据发生到下游， 如果没有使用out ,下游不在有数据
        */
      override def processElement(line: String, //一行数据
                                  ctx: ProcessFunction[String, String]#Context, //上下文对象
                                  out: Collector[String]): Unit = {

        val gender: String = line.split(",")(3)

        //side out
        gender match {
          case "男" => ctx.output(f, line)
          case "女" => ctx.output(m, line)
        }
      }
    })


    //通过tag获取数据
    val sideFDS: DataStream[String] = sideOutDS.getSideOutput(f)
    sideFDS.print()

    //val sideMDS: DataStream[String] = sideOutDS.getSideOutput(m)


    env.execute()


  }

}
